package edu.csl.study.spark.basic

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

//todo: 利用scala语言开发spark程序实现单词统计

/**
 *  Maven打包成jar后，到服务器执行：
 * spark-submit \
 * --master spark://centos20:7077 \
 * --class edu.csl.study.spark.basic.WordCount_Scala_Spark \
 * --executor-memory 1g  \
 * --total-executor-cores 2 \
 * /home/coreapp/spark-2.4.5-bin-without-hadoop/examples/my/StudySpark-1.0-SNAPSHOT.jar \
 * /words.txt  /words_out
 *
 */
object Core_WordCount_Scala_Cluster {
  def main(args: Array[String]): Unit = {
    //1、构建sparkConf对象 设置application名称
    val sparkConf: SparkConf = new SparkConf().setAppName("WordCountOnSpark")

    //2、构建sparkContext对象,该对象非常重要，它是所有spark程序的执行入口
    // 它内部会构建  DAGScheduler和 TaskScheduler 对象
    val sc = new SparkContext(sparkConf)

    //设置日志输出级别
    sc.setLogLevel("warn")

    //3、读取数据文件
    val data: RDD[String] = sc.textFile(args(0))

    //4、 切分每一行，获取所有单词
    val words: RDD[String] = data.flatMap(x=>x.split(" "))

    //5、每个单词计为1
    val wordAndOne: RDD[(String, Int)] = words.map(x => (x,1))

    //6、相同单词出现的1累加
    val result: RDD[(String, Int)] = wordAndOne.reduceByKey((x,y)=>x+y)

    //7、把计算结果保存在hdfs上
    result.saveAsTextFile(args(1)+new SimpleDateFormat("yyyyMMddHHmmss").format(new Date()))

    //8、关闭sc
    sc.stop()

  }


}